package com.ycz.consumer;

import com.rabbitmq.client.*;
import com.ycz.config.RabbitConnectionUtil;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @author ycz
 * @date 2021/6/11
 *
 */
@Configuration
public class ConsumerTest {

    static String queueName = "hello";
    String exchangeName = "helloExchange";
    String topicExchange = "topicExchange";

    public void consumerByFanout() {
        try {
            Connection conn = RabbitConnectionUtil.getRabbitConnection();
            Channel channel = conn.createChannel();
            channel.exchangeDeclare(exchangeName, "fanout"); // 声明fanout交换器
            // channel.queueDeclare().getQueue()为随机队列，Rabbit会随机生成队列名称，一旦消费者断开连接，该队列会自动删除
            String queueName = channel.queueDeclare().getQueue(); // 声明队列
            // TODO
            // fanout和direct的区别最多的在接收端，fanout需要绑定队列到对应的交换器用于订阅消息。
            channel.queueBind(queueName, exchangeName, "");
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");

                    System.out.println("consumerByFanout 消息正文：" + message);
                    System.out.println("queueName ：" + queueName);
                }
            };
            channel.basicConsume(queueName, true, consumer);
        } catch (Exception e) {

        }
    }

    public void consumerByTopic() {
        try {
            Connection conn = RabbitConnectionUtil.getRabbitConnection();
            Channel channel = conn.createChannel();
            channel.exchangeDeclare(topicExchange, "topic"); // 声明topic交换器
            // channel.queueDeclare().getQueue()为随机队列，Rabbit会随机生成队列名称，一旦消费者断开连接，该队列会自动删除
            String queueName = channel.queueDeclare().getQueue(); // 声明队列
            // 路由键匹配
            String routingKey = "#.error";
            channel.queueBind(queueName, topicExchange, routingKey);
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");

                    System.out.println("consumerByTopic 消息正文：" + message);
                    System.out.println("consumerByTopic ：" + queueName);
                }
            };
            channel.basicConsume(queueName, true, consumer);
        } catch (Exception e) {

        }
    }

    public void consumer() {
        Connection connection = RabbitConnectionUtil.getRabbitConnection();
        try {
            // 创建通道
            Channel channel = connection.createChannel();
            // 声明队列【参数说明：参数一：队列名称，参数二：是否持久化；参数三：是否独占模式；参数四：消费者断开连接时是否删除队列；参数五：消息其他参数】
            channel.queueDeclare(queueName, true, false, false, null);
            // 创建订阅器 并接收消息
            // ==== 消息确认
            // confirm(channel);
            // confirm2(channel);
            // 消息在确认之前，可以有两个选择：
            // 选择1：断开与Rabbit的连接，这样Rabbit会重新把消息分派给另一个消费者；
            // 选择2：拒绝Rabbit发送的消息使用channel.basicReject(long deliveryTag, boolean requeue)，参数1：消息的id；参数2：处理消息的方式，如果是true，Rabbib会重新分配这个消息给其他订阅者，如果设置成false的话，Rabbit会把消息发送到一个特殊的“死信”队列，用来存放被拒绝而不重新放入队列的消息。
            reject(channel);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void reject(Channel channel) throws IOException {
        GetResponse resp = channel.basicGet(queueName, false);
        String message = new String(resp.getBody(), "utf-8");
        // 拒绝Rabbit发送的消息使用channel.basicReject(long deliveryTag, boolean requeue)，参数1：消息的id；参数2：处理消息的方式，
        // 如果是true，Rabbib会重新分配这个消息给其他订阅者，如果设置成false的话，Rabbit会把消息发送到一个特殊的“死信”队列，用来存放被拒绝而不重新放入队列的消息。
        System.out.println("消息id：" + resp.getEnvelope());
        System.out.println("消息正文：" + message);
        channel.basicReject(resp.getEnvelope().getDeliveryTag(), true);
        System.out.println("消息id：" + resp.getEnvelope() + "已拒绝接收");
    }

    public void confirm(Channel channel) throws IOException {
        channel.basicConsume(queueName, false, "", new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 队列名称
                String routingKey = envelope.getRoutingKey();
                String contentType = properties.getContentType();
                String content = new String(body, "utf-8");
                System.out.println("消息正文：" + content);
                System.out.printf("routingKey:", routingKey);
                System.out.printf("contentType:", contentType);
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }

    public void confirm2(Channel channel) throws IOException {
        GetResponse resp = channel.basicGet(queueName, false);
        String message = new String(resp.getBody(), "utf-8");
        System.out.println("消息id：" + resp.getEnvelope());
        System.out.println("消息正文：" + message);
        // 消息id，true:批量确认小于id的消息
        channel.basicAck(resp.getEnvelope().getDeliveryTag(), false);
    }

}
